package ynu.edu.item.listener;

import com.fasterxml.jackson.core.type.TypeReference;
import com.rabbitmq.client.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import ynu.edu.common.constant.RedisConstants;
import ynu.edu.common.domain.stats.CategoryStat;
import ynu.edu.common.domain.stats.StatsData;
import ynu.edu.common.exception.RedisOperationException;
import ynu.edu.common.handler.RetryHandler;
import ynu.edu.common.utils.RedisCache;
import ynu.edu.item.mapper.ItemMapper;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Component
@RequiredArgsConstructor
public class ItemStatsMessageListener {

    private final RedisCache redisCache;
    private final ItemMapper itemMapper;

    @Value("${spring.rabbitmq.retry.max-attempts:3}")
    private int maxRetries;

    @RabbitListener(queues = "${rabbitmq.queue-name}", containerFactory = "rabbitListenerContainerFactory")
    public void onMessage(String message, Channel channel,
                          @Header(AmqpHeaders.DELIVERY_TAG) long tag,
                          @Header(name = "x-delivery-count", required = false) Integer deliveryCount) {
        log.info("[StatsListener] 收到消息: {}", message);

        boolean success = false;
        try {
            // 1. 查询统计信息
            List<CategoryStat> categoryStats = itemMapper.selectCategoryStats();
            Long totalCount = itemMapper.selectTotalCount();

            // 2. 构建统计数据
            StatsData<List<CategoryStat>> statsData = StatsData.success(categoryStats, totalCount);

            // 3. 写入 Redis
            redisCache.setCacheObject(RedisConstants.ITEM_STATS_KEY, statsData, 24, TimeUnit.HOURS);

            // 4. 显式校验写入结果
            StatsData<List<CategoryStat>> cachedObj = redisCache.getCacheObject(
                    RedisConstants.ITEM_STATS_KEY,
                    new TypeReference<>() {}
            );
            if (cachedObj == null) {
                throw new RedisOperationException("Redis 数据不存在");
            }
            if (!cachedObj.equals(statsData)) {
                throw new RedisOperationException("Redis 数据校验失败");
            }

            log.info("[StatsListener] 分类统计更新成功，总数: {}", totalCount);
            success = true;

        } catch (RedisOperationException e) {
            log.error("[StatsListener] Redis操作失败", e);
            boolean shouldRequeue = RetryHandler.shouldRetry(deliveryCount, maxRetries, e);
            handleNack(channel, tag, shouldRequeue);
        } catch (Exception e) {
            log.error("[StatsListener] 未知异常，进入死信队列", e);
        } finally {
            if (success) {
                handleAck(channel, tag);
            } else {
                handleNack(channel, tag, false);
            }
        }
    }

    // 统一处理消息确认（Ack）
    private void handleAck(Channel channel, long tag) {
        try {
            if (channel.isOpen()) {
                channel.basicAck(tag, false);
                log.debug("[StatsListener] 消息确认成功, tag: {}", tag);
            } else {
                log.warn("[StatsListener] 通道已关闭，无法确认消息, tag: {}", tag);
            }
        } catch (IOException e) {
            log.error("[StatsListener] 确认消息失败, tag: {}", tag, e);
        }
    }

    // 统一处理消息拒绝（Nack）
    private void handleNack(Channel channel, long tag, boolean requeue) {
        try {
            if (channel.isOpen()) {
                channel.basicNack(tag, false, requeue);
                log.debug("[StatsListener] 消息拒绝成功, tag: {}, requeue: {}", tag, requeue);
            } else {
                log.warn("[StatsListener] 通道已关闭，无法拒绝消息, tag: {}", tag);
            }
        } catch (IOException e) {
            log.error("[StatsListener] 拒绝消息失败, tag: {}", tag, e);
        }
    }
}

